package com.tianji.learning.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tianji.api.client.course.CourseClient;
import com.tianji.api.dto.course.CourseFullInfoDTO;
import com.tianji.api.dto.leanring.LearningLessonDTO;
import com.tianji.api.dto.leanring.LearningRecordDTO;
import com.tianji.api.dto.leanring.LearningRecordFormDTO;
import com.tianji.common.constants.RedisConstants;
import com.tianji.common.utils.*;
import com.tianji.learning.entity.LearningLesson;
import com.tianji.learning.entity.LearningRecord;
import com.tianji.learning.mapper.LearningRecordMapper;
import com.tianji.learning.pojo.DelayedTask;
import com.tianji.learning.pojo.SectionType;
import com.tianji.learning.service.ILearningLessonService;
import com.tianji.learning.service.ILearningRecordService;
import com.tianji.learning.vo.LearningLessonVO;
import com.tianji.learning.vo.LessonStatus;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 * 学习记录表 服务实现类
 * </p>
 *
 * @author xjs
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class LearningRecordServiceImpl extends ServiceImpl<LearningRecordMapper, LearningRecord> implements ILearningRecordService {

    private final ILearningLessonService lessonService;

    private final CourseClient courseClient;

    private final StringRedisTemplate redisTemplate;

    private final PlatformTransactionManager transactionManager;

    private DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();

//    public LearningRecordServiceImpl(ILearningLessonService lessonService){
//        this.lessonService = lessonService;
//    }

    /**
     * 为了系统启动的时候，异步的从队列中读取消息
     * */
    private ThreadPoolExecutor takeTaskPool = new ThreadPoolExecutor(1, 1, 120, TimeUnit.SECONDS,  new ArrayBlockingQueue<>(1));

    /**
     * 为了执行任务，因为单个任务执行时间可能会比较长
     * */
    private ThreadPoolExecutor executeTaskPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors(), 120, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(Runtime.getRuntime().availableProcessors()),
            new ThreadPoolExecutor.CallerRunsPolicy());

    /**
     * spring生命周期回调，读取delayQueue
     * */
    @PostConstruct
    public void postConstruct(){
        takeTaskPool.submit(new Runnable() {
            @Override
            public void run() {
                while(true){
                    try{
                        // 读取delayQueue中消息，没有到期的会等待
                        DelayedTask msg = delayQueue.take();
                        // 为了防止业务逻辑执行太长，阻塞消息的读取
                        executeTaskPool.submit(new Runnable() {
                            @Override
                            public void run() {
                                // 业务逻辑，把缓存的信息同步到数据库中
                                syncRedisToMysql((LearningRecord)msg.getData());
                            }
                        });
                    }catch(Exception e){
                        log.error(e.getMessage());
                    }
                }
            }
        });
    }

    private void syncRedisToMysql(LearningRecord msg) {
        log.info("从延迟队列读取消息，lessonId：{}， sectionId：{}, msg中的moment:{}", msg.getLessonId(), msg.getSectionId(), msg.getMoment());
        // 判断msg中的进度和redis中的进度是否是一致
        // 从redis读取最新的进度
        String key = getLearningRecordKey(msg.getLessonId(), msg.getUserId());
        String json = (String)redisTemplate.opsForHash().get(key, msg.getSectionId() + "");
        if(StringUtils.isEmpty(json)){
            return;
        }
        LearningRecord redisRecord = JsonUtils.toBean(json, LearningRecord.class);
        log.info("从redis读取学习记录，lessonId：{}， sectionId：{}, redis中的moment:{}", redisRecord.getLessonId(), redisRecord.getSectionId(), redisRecord.getMoment());
        // 如果不一致，说明用户还在看，不用管
        if(!Objects.equals(redisRecord.getMoment(),msg.getMoment())){
            log.info("视频持续播放中，lessonId：{}， sectionId：{}, msg中的moment：{} redis中的moment:{}",
                    redisRecord.getLessonId(), redisRecord.getSectionId(), msg.getMoment(), redisRecord.getMoment());
            return;
        }
        log.info("视频已经退出播放，lessonId：{}， sectionId：{}, moment:{}", redisRecord.getLessonId(), redisRecord.getSectionId(), redisRecord.getMoment());
        TransactionDefinition definition = TransactionDefinition.withDefaults();
        // 开启事务
        TransactionStatus transactionStatus = transactionManager.getTransaction(definition);
        try{
            // 如果是一致，说明用户已经不看了
            // 更新学习记录，主要是更新进度
            updateLearningRecord(redisRecord.getId(), redisRecord.getMoment(), false);
            //   更新课表，更新课表中最近学习的小节和学习时间
            updateLessonLatestInfo(redisRecord.getLessonId(), redisRecord.getSectionId());
            // 提交事务
            transactionManager.commit(transactionStatus);
        }catch(Exception e){
            log.error(ExceptionUtils.getStackTrace(e));
            // 回滚事务
            transactionManager.rollback(transactionStatus);
        }
    }


    @Override
    public int getLearnedSectionCountByUserIdFinishTime(Long userId, Long lessonId, LocalDateTime weekBeginTime, LocalDateTime weekEndTime) {
        LambdaQueryWrapper<LearningRecord> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(LearningRecord::getUserId, userId)
                .eq(LearningRecord::getFinished, true)
                .eq(lessonId != null,LearningRecord::getLessonId, lessonId)
                .gt(weekBeginTime!=null,LearningRecord::getFinishTime, weekBeginTime)
                .lt(weekEndTime != null, LearningRecord::getFinishTime, weekEndTime);
        return this.count(queryWrapper);
    }

    private void insertLearningRecord(LearningRecordFormDTO dto, Long userId, boolean finished){
        LearningRecord learningRecord = BeanUtils.copyBean(dto, LearningRecord.class);
        learningRecord.setUserId(userId);
        if(finished){
            learningRecord.setFinished(true);
            learningRecord.setFinishTime(LocalDateTime.now());
        }
        learningRecord.setCreateTime(LocalDateTime.now());
        learningRecord.setUpdateTime(LocalDateTime.now());
        this.save(learningRecord);
    }

    private void updateLearningRecord(Long recordId, Integer moment, boolean finished){
        LearningRecord recordUpdate = new LearningRecord();
        recordUpdate.setId(recordId);
        recordUpdate.setMoment(moment);
        recordUpdate.setUpdateTime(LocalDateTime.now());
        if(finished){
            recordUpdate.setFinished(true);
            recordUpdate.setFinishTime(LocalDateTime.now());
        }
        this.updateById(recordUpdate);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void submitLearningRecord(LearningRecordFormDTO dto) {
        // 先获取用户id
        Long userId = UserContext.getUser();
        // 先判断是否是考试
        Integer sectionType = dto.getSectionType();
        // 如果是考试
        if(Objects.equals(sectionType, SectionType.EXAM.getValue())){
            //   新增一条学习记录，并且已完成
            insertLearningRecord(dto, userId, true);
            //   课表中已学习的小节数量+1， 如果课表的状态是未学习，改成已学习
            updateLessonLearnedSectionAndStatus(dto.getLessonId());
        }else{
            // 如果是视频
            // 先查询是否存在学习记录，先查redis，再查数据库
            LearningRecord recordDB = queryLearningRecord(dto.getLessonId(), dto.getSectionId(), userId);
            if(recordDB == null){
                // 如果不存在 ，新增学习记录insert
                insertLearningRecord(dto, userId, false);
                // 更新课表中最近学习的小节和学习时间，如果是未学习，改成学习中
                updateLessonLatestInfo(dto.getLessonId(), dto.getSectionId());
            }else{
                // 判断是否是第一次学完，进度大于50% && finished=false
                // 播放进度
                Integer moment = dto.getMoment();
                // 视频总时长
                Integer duration = dto.getDuration();
                // 是否超过50%
                boolean moreThan50 = moment*2>duration;
                // 数据库中之前的状态是否是已经完成
                boolean finishDB = recordDB.getFinished();
                //   是否是第一次学完，进度大于50% && finished=false
                boolean firstFinish = moreThan50 && !finishDB;
                if(firstFinish){
                    log.info("首次学完当前小节，lessonId：{}， sectionId：{}", dto.getLessonId(), dto.getSectionId());
                    // 更新学习记录
                    updateLearningRecord(recordDB.getId(), moment, true);
                    // 删除redis的缓存，因为finish状态边为已完成，进度信息也发生了变更
                    String key = getLearningRecordKey(dto.getLessonId(), userId);
                    redisTemplate.opsForHash().delete(key, ""+dto.getSectionId());
                    // 是，已学习小节数量+1， 判断已学习的小节数量 >= 课程的总的小节数量，更新状态是已学完
                    updateLessonLearnedSectionAndStatus(dto.getLessonId());
                }else{
                    log.info("不是首次学完当前小节，lessonId：{}， sectionId：{}, 旧的moment:{}", dto.getLessonId(), dto.getSectionId(), recordDB.getMoment());
                    // 把进度信息写入redis，不写数据库
                    String key = getLearningRecordKey(dto.getLessonId(), userId);
                    recordDB.setMoment(dto.getMoment());
                    recordDB.setUpdateTime(LocalDateTime.now());
                    redisTemplate.opsForHash().put(key, ""+dto.getSectionId(), JsonUtils.toJsonStr(recordDB));
                    log.info("更新缓存中的进度，lessonId：{}， sectionId：{}, 新的moment:{}", dto.getLessonId(), dto.getSectionId(), dto.getMoment());
                    // 发延迟20秒的延迟任务，结束
                    DelayedTask delayedTask = new DelayedTask(recordDB,20);
                    delayQueue.put(delayedTask);
                    log.info("发送延迟消息，lessonId：{}， sectionId：{}, moment:{}", dto.getLessonId(), dto.getSectionId(), dto.getMoment());
                    // 接下来的处理逻辑：写一个消费者，监听任务队列，收到消息之后，判断消息中的进度和redis中最新的进度是否一致，
                    //     如果是一致，说明用户已经不看视频了，就可以把缓存的数据写入数据库
                    //     如果是不一致，说明用户还在持续的观看视频，不需要做处理
                }
            }
        }
    }

    private String getLearningRecordKey(Long lessonId, Long userId){
        return RedisConstants.LEARNING_RECORD_KEY+ "" + userId + "_" + lessonId;
    }

    private LearningRecord queryLearningRecord(Long lessonId, Long sectionId, Long userId) {
        String key = getLearningRecordKey(lessonId, userId);
        // 先查缓存
        log.info("查询缓存中的学习记录，lessonId：{}， sectionId：{}", lessonId, sectionId);
        String json = (String)redisTemplate.opsForHash().get(key, "" + sectionId);
        // 缓存有，直接返回
        if(!StringUtils.isEmpty(json)){
            LearningRecord record = JsonUtils.toBean(json, LearningRecord.class);
            log.info("缓存命中，lessonId：{}， sectionId：{}, moment:{}", lessonId, sectionId, record.getMoment());
            return record;
        }
        log.info("缓存未命中，查询数据库，lessonId：{}， sectionId：{}", lessonId, sectionId);
        // 缓存没有，查数据库
        LambdaQueryWrapper<LearningRecord> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(LearningRecord::getLessonId, lessonId)
                .eq(LearningRecord::getSectionId, sectionId)
                .eq(LearningRecord::getUserId, userId);
        LearningRecord recordDB = this.getOne(queryWrapper);
        // 如果数据不为空，写入redis
        if(recordDB != null){
            log.info("缓存未命中，查询到数据，lessonId：{}， sectionId：{}, moment:{} ", lessonId, sectionId, recordDB.getMoment());
            redisTemplate.opsForHash().put(key, ""+sectionId, JsonUtils.toJsonStr(recordDB));
        }
        return recordDB;
    }

    /***
     * 更新课表中的已学习的小节数量和学习状态
     */
    private void updateLessonLearnedSectionAndStatus(Long lessonId){
        // 查询课表
        LearningLesson lesson = lessonService.getById(lessonId);
        AssertUtils.isNotNull(lesson,"查询不到课表信息：" + lessonId);
        LearningLesson lessonUpdate = new LearningLesson();
        lessonUpdate.setId(lessonId);
        // 课表中已学习的小节数量+1
        lessonUpdate.setLearnedSections(lesson.getLearnedSections() + 1);
        // 如果课表的状态是未学习，改成已学习
        if(Objects.equals(lesson.getStatus(), LessonStatus.NOT_BEGIN.getValue())){
            lessonUpdate.setStatus(LessonStatus.LEARNING.getValue());
        }
        //   查询课程的总的小节数量
        //   判断已学习的小节数量 >= 课程的总的小节数量
        //   是，更新课表的状态为已经学完
        CourseFullInfoDTO course = courseClient.getCourseInfoById(lesson.getCourseId(), false, false);
        AssertUtils.isNotNull(course,"查询不到课程信息：" + lesson.getCourseId());
        // 总的小节数量
        Integer sectionNum = course.getSectionNum();
        if(lesson.getLearnedSections() + 1 >= sectionNum){
            lessonUpdate.setStatus(LessonStatus.FINISHED.getValue());
        }
        lessonService.updateById(lessonUpdate);
    }

    /**
     * 更新课表中的最近的学习的小节ID和最近的学习时间
     * */
    private void updateLessonLatestInfo(Long lessonId, Long sectionId){
        LearningLesson lesson = lessonService.getById(lessonId);
        AssertUtils.isNotNull(lesson,"查询不到课表信息：" + lessonId);
        // 更新
        LearningLesson lessonUpdate = new LearningLesson();
        lessonUpdate.setId(lessonId);
        lessonUpdate.setLatestSectionId(sectionId);
        lessonUpdate.setLatestLearnTime(LocalDateTime.now());
        if(Objects.equals(lesson.getStatus(), LessonStatus.NOT_BEGIN.getValue())){
            lessonUpdate.setStatus(LessonStatus.LEARNING.getValue());
        }
        lessonService.updateById(lessonUpdate);
    }

    @Override
    public LearningLessonDTO queryLearningRecordByCourse(Long courseId) {
        // 先获取userId
        Long userId = UserContext.getUser();
        // 先根据courseId和userId查课表，得到课表的id 最近学习的小节
        LearningLessonVO lesson = lessonService.getLessonByUserIdCourseId(userId, courseId);
        if(lesson == null){
            return null;
        }
        // 根据lesson_id查询学习记录，得到用户的学习记录列表
        LambdaQueryWrapper<LearningRecord> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(LearningRecord::getLessonId, lesson.getId());
        List<LearningRecord> recordList = this.list(queryWrapper);
        // 拼装输出结果
        LearningLessonDTO result = new LearningLessonDTO();
        result.setId(lesson.getId());
        result.setLatestSectionId(lesson.getLatestSectionId());
        result.setRecords(BeanUtils.copyList(recordList, LearningRecordDTO.class));
        return result;
    }
}
